package dispatcher

import (
	"math"
	"reflect"
	"regexp"
	"sync/atomic"
	"time"

	cb "git.cloud.inspur.com/ichain/ichain-protos-go/common"
	pb "git.cloud.inspur.com/ichain/ichain-protos-go/inode"
	"git.cloud.inspur.com/ichain/ichain-sdk-go/pkg/common/logging"
	"git.cloud.inspur.com/ichain/ichain-sdk-go/thirdparty/ichain/protoutil"
	"github.com/pkg/errors"
)

var logger = logging.NewLogger("ichainsdk/ichain")

const (
	dispatcherStateInitial = iota
	dispatcherStateStarted
	dispatcherStateStopped
)

const (
	eventConsumerBufferSize = 100
	eventConsumerTimeout    = 500 * time.Millisecond
)

type Handler func(Event)

// Dispatcher
// @Description: 负责连接event服务以及管理所有订阅事件
type Dispatcher struct {
	latestBlockNum             uint64 // Must be first, do not move
	updateLastBlockInfoOnly    bool
	state                      int32
	eventch                    chan interface{}
	blockRegistrations         []*BlockReg              // 所有区块注册
	filteredBlockRegistrations []*FilteredBlockReg      // 过滤区块信息的注册
	handlers                   map[reflect.Type]Handler // 所有回调函数
	txRegistrations            map[string]*TxStatusReg  // 所有交易注册
	ccRegistrations            map[string]*ContractReg  // 所有合约注册
}

func New() *Dispatcher {
	logger.Debug("Creating new dispatcher.")

	return &Dispatcher{
		handlers:        make(map[reflect.Type]Handler),
		eventch:         make(chan interface{}, eventConsumerBufferSize),
		txRegistrations: make(map[string]*TxStatusReg),
		ccRegistrations: make(map[string]*ContractReg),
		state:           dispatcherStateInitial,
		latestBlockNum:  math.MaxUint64,
	}
}

// RegisterHandlers  注册所有事件类型
func (ed *Dispatcher) RegisterHandlers() {
	ed.RegisterHandler(&RegisterContractEvent{}, ed.handleRegisterCCEvent)
	ed.RegisterHandler(&RegisterTxStatusEvent{}, ed.handleRegisterTxStatusEvent)
	ed.RegisterHandler(&RegisterBlockEvent{}, ed.handleRegisterBlockEvent)
	ed.RegisterHandler(&RegisterFilteredBlockEvent{}, ed.handleRegisterFilteredBlockEvent)
	ed.RegisterHandler(&UnregisterEvent{}, ed.handleUnregisterEvent)
	ed.RegisterHandler(&StopEvent{}, ed.HandleStopEvent)
	ed.RegisterHandler(&RegistrationInfoEvent{}, ed.handleRegistrationInfoEvent)
}

func (ed *Dispatcher) EventCh() (chan<- interface{}, error) {
	state := ed.getState()
	if state == dispatcherStateStarted {
		return ed.eventch, nil
	}
	return nil, errors.Errorf("dispatcher not started - Current state [%d]", state)
}

// Start 初始化所有注册函数，监听传入的事件
func (ed *Dispatcher) Start() error {
	logger.Debugf("dispatcher start")
	if !ed.setState(dispatcherStateInitial, dispatcherStateStarted) {
		return errors.New("cannot start dispatcher since it's not in its initial state")
	}

	ed.RegisterHandlers()

	go func() {
		for {
			if ed.getState() == dispatcherStateStopped {
				break
			}

			logger.Debug("Listening for events...")

			// 监听事件
			e, ok := <-ed.eventch
			if !ok {
				break
			}

			logger.Debugf("Received event: %+v", reflect.TypeOf(e))

			// 执行注册函数
			if handler, ok := ed.handlers[reflect.TypeOf(e)]; ok {
				logger.Debugf("Dispatching event: %+v", reflect.TypeOf(e))
				handler(e)
			} else {
				logger.Errorf("Handler not found for: %s", reflect.TypeOf(e))
			}
		}
		logger.Debug("Exiting event dispatcher")
	}()
	return nil
}

// LatestBlockNum 最新的区块高度
// @return uint64
func (ed *Dispatcher) LatestBlockNum() uint64 {
	return atomic.LoadUint64(&ed.latestBlockNum)
}

// updateLatestBlockNum 更新最新区块高度
// @param blockNum
// @return error
func (ed *Dispatcher) updateLatestBlockNum(blockNum uint64) error {
	latestBlockNum := atomic.LoadUint64(&ed.latestBlockNum)
	if latestBlockNum == math.MaxUint64 || blockNum > latestBlockNum {
		atomic.StoreUint64(&ed.latestBlockNum, blockNum)
		logger.Debugf("Updated last block received to %d", blockNum)
		return nil
	}
	return errors.Errorf("Expecting a block number greater than %d but received block number %d", latestBlockNum, blockNum)
}

// clearRegistrations 清理所有注册函数
// @param closeChain
func (ed *Dispatcher) clearRegistrations(closeChain bool) {
	ed.clearBlockRegistrations(closeChain)
	ed.clearFilteredBlockRegistrations(closeChain)
	ed.clearTxRegistrations(closeChain)
	ed.clearContractRegistrations(closeChain)
}

// clearBlockRegistrations 清理区块的注册函数
// @param closeChain
func (ed *Dispatcher) clearBlockRegistrations(closeChain bool) {
	if closeChain {
		for _, reg := range ed.blockRegistrations {
			close(reg.Eventch)
		}
	}
	ed.blockRegistrations = nil
}

// clearFilteredBlockRegistrations 清理过滤区块的注册函数
// @param closeChain
func (ed *Dispatcher) clearFilteredBlockRegistrations(closeChain bool) {
	if closeChain {
		for _, reg := range ed.filteredBlockRegistrations {
			close(reg.Eventch)
		}
	}
	ed.filteredBlockRegistrations = nil
}

// clearTxRegistrations 清理交易注册函数
// @param closeChain
func (ed *Dispatcher) clearTxRegistrations(closeChain bool) {
	if closeChain {
		for _, reg := range ed.txRegistrations {
			logger.Debugf("Closing TX registration event chain for TxID [%s].", reg.TxID)
			close(reg.Eventch)
		}
	}
	ed.txRegistrations = make(map[string]*TxStatusReg)
}

// clearContractRegistrations 清理合约注册函数
// @param closeChain
func (ed *Dispatcher) clearContractRegistrations(closeChain bool) {
	if closeChain {
		for _, reg := range ed.ccRegistrations {
			logger.Debugf("Closing contract registration event chain for CC ID [%s] and event filter [%s].", reg.ContractID, reg.EventFilter)
			close(reg.Eventch)
		}
	}
	ed.ccRegistrations = make(map[string]*ContractReg)
}

func (ed *Dispatcher) HandleStopEvent(e Event) {
	event := e.(*StopEvent)

	logger.Debugf("Stopping dispatcher...")
	if !ed.setState(dispatcherStateStarted, dispatcherStateStopped) {
		logger.Warn("Cannot stop event dispatcher since it's already stopped.")
		event.ErrCh <- errors.New("dispatcher already stopped")
		return
	}

	ed.clearRegistrations(true)

	event.ErrCh <- nil
}

func (ed *Dispatcher) handleRegisterBlockEvent(e Event) {
	event := e.(*RegisterBlockEvent)

	ed.registerBlockEvent(event.Reg)
	event.RegCh <- event.Reg
}

func (ed *Dispatcher) registerBlockEvent(reg *BlockReg) {
	ed.blockRegistrations = append(ed.blockRegistrations, reg)
}

func (ed *Dispatcher) handleRegisterFilteredBlockEvent(e Event) {
	event := e.(*RegisterFilteredBlockEvent)
	ed.registerFilteredBlockEvent(event.Reg)
	event.RegCh <- event.Reg
}

func (ed *Dispatcher) registerFilteredBlockEvent(reg *FilteredBlockReg) {
	ed.filteredBlockRegistrations = append(ed.filteredBlockRegistrations, reg)
}

func (ed *Dispatcher) handleRegisterCCEvent(e Event) {
	event := e.(*RegisterContractEvent)

	regExp, err := regexp.Compile(event.Reg.EventFilter)
	if err != nil {
		event.ErrCh <- errors.Wrapf(err, "error compiling regular expression for event filter [%s]", event.Reg.EventFilter)
	} else {
		event.Reg.EventRegExp = regExp
		if err := ed.registerCCEvent(event.Reg); err != nil {
			event.ErrCh <- err
		} else {
			event.RegCh <- event.Reg
		}
	}
}

func (ed *Dispatcher) registerCCEvent(reg *ContractReg) error {
	key := getCCKey(reg.ContractID, reg.EventFilter)
	if _, exists := ed.ccRegistrations[key]; exists {
		return errors.Errorf("registration already exists for contract [%s] and event [%s]", reg.ContractID, reg.EventFilter)
	}
	ed.ccRegistrations[key] = reg
	return nil
}

func (ed *Dispatcher) handleRegisterTxStatusEvent(e Event) {
	event := e.(*RegisterTxStatusEvent)

	if err := ed.registerTxStatusEvent(event.Reg); err != nil {
		event.ErrCh <- err
	} else {
		event.RegCh <- event.Reg
	}
}

func (ed *Dispatcher) registerTxStatusEvent(reg *TxStatusReg) error {
	if _, exists := ed.txRegistrations[reg.TxID]; exists {
		return errors.Errorf("registration already exists for TX ID [%s]", reg.TxID)
	}
	ed.txRegistrations[reg.TxID] = reg
	return nil
}

func (ed *Dispatcher) handleUnregisterEvent(e Event) {
	event := e.(*UnregisterEvent)

	var err error
	switch registration := event.Reg.(type) {
	case *BlockReg:
		err = ed.unregisterBlockEvents(registration)
	case *FilteredBlockReg:
		err = ed.unregisterFilteredBlockEvents(registration)
	case *ContractReg:
		err = ed.unregisterCCEvents(registration)
	case *TxStatusReg:
		err = ed.unregisterTXEvents(registration)
	default:
		err = errors.Errorf("Unsupported registration type: %+v", reflect.TypeOf(registration))
	}
	if err != nil {
		logger.Warnf("Error in unregister: %s", err)
	}
}

func (ed *Dispatcher) handleRegistrationInfoEvent(e Event) {
	evt := e.(*RegistrationInfoEvent)

	regInfo := &RegistrationInfo{
		NumBlockRegistrations:         len(ed.blockRegistrations),
		NumFilteredBlockRegistrations: len(ed.filteredBlockRegistrations),
		NumCCRegistrations:            len(ed.ccRegistrations),
		NumTxStatusRegistrations:      len(ed.txRegistrations),
	}

	regInfo.TotalRegistrations =
		regInfo.NumBlockRegistrations + regInfo.NumFilteredBlockRegistrations + regInfo.NumCCRegistrations + regInfo.NumTxStatusRegistrations

	evt.RegInfoCh <- regInfo
}

// HandleBlock 处理区块事件
func (ed *Dispatcher) HandleBlock(block *cb.Block, sourceURL string) {
	logger.Infof("Handling block event - Block #%d", block.Header.Number)

	// 更新最新的区块高度
	if err := ed.updateLatestBlockNum(block.Header.Number); err != nil {
		logger.Error(err.Error())
		return
	}

	if ed.updateLastBlockInfoOnly {
		ed.updateLastBlockInfoOnly = false
		return
	}

	logger.Info("Publishing block event...")
	logger.Infof("Handling block event - Block #%d", block.Header.Number)
	ed.publishBlockEvents(block, sourceURL)
	ed.publishFilteredBlockEvents(protoutil.ToFilteredBlock(block), sourceURL)
}

func (ed *Dispatcher) HandleFilteredBlock(fblock *pb.FilteredBlock, sourceURL string) {
	logger.Infof("Handling filtered block event - Block #%d", fblock.Number)

	if err := ed.updateLatestBlockNum(fblock.Number); err != nil {
		logger.Error(err.Error())
		return
	}

	if ed.updateLastBlockInfoOnly {
		ed.updateLastBlockInfoOnly = false
		return
	}

	logger.Debug("Publishing filtered block event...")
	ed.publishFilteredBlockEvents(fblock, sourceURL)
}

func (ed *Dispatcher) unregisterBlockEvents(registration *BlockReg) error {
	for i, reg := range ed.blockRegistrations {
		if reg == registration {
			// Move the 0'th item to i and then delete the 0'th item
			ed.blockRegistrations[i] = ed.blockRegistrations[0]
			ed.blockRegistrations = ed.blockRegistrations[1:]
			close(reg.Eventch)
			return nil
		}
	}
	return errors.New("the provided registration is invalid")
}

func (ed *Dispatcher) unregisterFilteredBlockEvents(registration *FilteredBlockReg) error {
	for i, reg := range ed.filteredBlockRegistrations {
		if reg == registration {
			// Move the 0'th item to i and then delete the 0'th item
			ed.filteredBlockRegistrations[i] = ed.filteredBlockRegistrations[0]
			ed.filteredBlockRegistrations = ed.filteredBlockRegistrations[1:]
			close(reg.Eventch)
			return nil
		}
	}
	return errors.New("the provided registration is invalid")
}

func (ed *Dispatcher) unregisterCCEvents(registration *ContractReg) error {
	key := getCCKey(registration.ContractID, registration.EventFilter)
	reg, ok := ed.ccRegistrations[key]
	if !ok {
		return errors.New("the provided registration is invalid")
	}

	logger.Debugf("Unregistering CC event for CC ID [%s] and event filter [%s]...", registration.ContractID, registration.EventFilter)
	close(reg.Eventch)
	delete(ed.ccRegistrations, key)
	return nil
}

func (ed *Dispatcher) unregisterTXEvents(registration *TxStatusReg) error {
	reg, ok := ed.txRegistrations[registration.TxID]
	if !ok {
		return errors.New("the provided registration is invalid")
	}

	logger.Debugf("Unregistering Tx Status event for TxID [%s]...", registration.TxID)
	close(reg.Eventch)
	delete(ed.txRegistrations, registration.TxID)
	return nil
}

func (ed *Dispatcher) publishBlockEvents(block *cb.Block, sourceURL string) {
	for _, reg := range ed.blockRegistrations {
		if !reg.Filter(block) {
			logger.Debugf("Not sending block event for block #%d since it was filtered out.", block.Header.Number)
			continue
		}
		select {
		case reg.Eventch <- NewBlockEvent(block, sourceURL):
		case <-time.After(eventConsumerTimeout):
			logger.Warn("Timed out sending block event.")
		}
	}
}

func (ed *Dispatcher) publishFilteredBlockEvents(fblock *pb.FilteredBlock, sourceURL string) {
	if fblock == nil {
		logger.Info("Filtered block is nil. Event will not be published")
		return
	}

	logger.Infof("Publishing filtered block event: %#v", fblock)
	checkFilteredBlockRegistrations(ed, fblock, sourceURL)

	for _, tx := range fblock.FilteredTransactions {
		ed.publishTxStatusEvents(tx, fblock.Number, sourceURL)

		if tx.TxValidationCode == pb.TxValidationCode_VALID {
			txActions := tx.GetTransactionActions()
			if txActions == nil {
				continue
			}
			logger.Infof("txActions.ContractActions: %v", txActions.ContractActions)
			if len(txActions.ContractActions) == 0 {
				logger.Infof("No contract action found for TxID[%s], block[%d], source URL[%s]", tx.Txid, fblock.Number, sourceURL)
			}
			for _, action := range txActions.ContractActions {
				if action.ContractEvent != nil {
					logger.Debugf("action.ContractEvent: evt=%s, tx=%s", action.ContractEvent.EventName, action.ContractEvent.TxId)
					ed.publishCCEvents(action.ContractEvent, fblock.Number, sourceURL)
				}
			}
		} else {
			logger.Infof("Cannot publish CCEvents for block[%d] and source URL[%s] since Tx Validation Code[%d] is not valid", fblock.Number, sourceURL, tx.TxValidationCode)
		}
	}
}

func checkFilteredBlockRegistrations(ed *Dispatcher, fblock *pb.FilteredBlock, sourceURL string) {
	for _, reg := range ed.filteredBlockRegistrations {
		logger.Debugf("consumerTimeout is %d >0", eventConsumerTimeout)
		select {
		case reg.Eventch <- NewFilteredBlockEvent(fblock, sourceURL):
		case <-time.After(eventConsumerTimeout):
			logger.Debugf("Timed out sending filtered block event.")
		}
	}
}

func (ed *Dispatcher) publishTxStatusEvents(tx *pb.FilteredTransaction, blockNum uint64, sourceURL string) {
	logger.Debugf("Publishing Tx Status event for TxID [%s]...", tx.Txid)
	if reg, ok := ed.txRegistrations[tx.Txid]; ok {
		logger.Debugf("Sending Tx Status event for TxID [%s] to registrant...", tx.Txid)
		select {
		case reg.Eventch <- NewTxStatusEvent(tx.Txid, tx.TxValidationCode, blockNum, sourceURL):
		case <-time.After(eventConsumerTimeout):
			logger.Warn("Timed out sending Tx Status event.")
			logger.Print("Timed out sending Tx Status event.")
		}
	}
}

func (ed *Dispatcher) publishCCEvents(ccEvent *pb.ContractEvent, blockNum uint64, sourceURL string) {
	for _, reg := range ed.ccRegistrations {
		logger.Infof("Matching CCEvent[%s,%s] against Reg[%s,%s] ...", ccEvent.ContractId, ccEvent.EventName, reg.ContractID, reg.EventFilter)
		if reg.ContractID == ccEvent.ContractId && reg.EventRegExp.MatchString(ccEvent.EventName) {
			logger.Debugf("... matched CCEvent[%s,%s] against Reg[%s,%s]", ccEvent.ContractId, ccEvent.EventName, reg.ContractID, reg.EventFilter)

			select {
			case reg.Eventch <- NewContractEvent(ccEvent.ContractId, ccEvent.EventName, ccEvent.TxId, ccEvent.Payload, blockNum, sourceURL):
			case <-time.After(eventConsumerTimeout):
				logger.Warn("Timed out sending CC event.")
			}
		}
	}
}

// RegisterHandler 注册函数
// @param t
// @param h
func (ed *Dispatcher) RegisterHandler(t interface{}, h Handler) {
	htype := reflect.TypeOf(t)
	if _, ok := ed.handlers[htype]; !ok {
		logger.Debugf("Registering handler for %s on dispatcher %T", htype, ed)
		ed.handlers[htype] = h
	} else {
		logger.Debugf("Cannot register handler %s on dispatcher %T since it's already registered", htype, ed)
	}
}

func (ed *Dispatcher) UpdateLastBlockInfoOnly() {
	ed.updateLastBlockInfoOnly = true
}

func getCCKey(ccID, eventFilter string) string {
	return ccID + "/" + eventFilter
}

func (ed *Dispatcher) getState() int32 {
	return atomic.LoadInt32(&ed.state)
}

func (ed *Dispatcher) setState(expectedState, newState int32) bool {
	return atomic.CompareAndSwapInt32(&ed.state, expectedState, newState)
}
